package com.atguigu.sparkcore.rdd.kvs

import com.atguigu.sparkcore.util.MySparkContextUtil
import org.apache.spark.{SparkConf, SparkContext}

/**
  * 通过key进行汇总，返回一个迭代器
  * author 剧情再美终是戏
  * mail 13286520398@163.com
  * date 2020/1/7 6:29 
  * version 1.0
  * <p>des</p>
  * <p>1、GroupByKey将相同key的值进行分组，直接shuffle,不会时行预聚合</p>
  * <p>2、源码：combineByKeyWithClassTag[CompactBuffer[V]](
  * createCombiner, mergeValue, mergeCombiners, partitioner, mapSideCombine = false)</p>
  * <p>3、GroupByKey必须在内存中持有所有的键值，所以当一个kev中有大量的value时，可能会导致OOM</p>
  * <p>4、性能不好，如果业务允许，优先使用aggregateByKey|reduceByKey，因为这两个转换算子都会有预聚合的操作</p>
  **/
object GroupByKey {

  def main(args: Array[String]): Unit = {

    // 获取sparkContext
    val sc = MySparkContextUtil.get(args)

    // 创建rdd
    val list = List(("female", 1), ("male", 5), ("male", 9), ("female", 5), ("male", 2), ("female", 15))
    val rdd = sc.makeRDD(list, 2)

    // 转换rdd
    val result = rdd.groupByKey()

    // 输出
    println(rdd.getNumPartitions)
    println(result.collect().mkString(","))

    // 关闭资源
    MySparkContextUtil.close(sc)
  }

}
